/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.writer;

import com.alibaba.fastjson.JSON;
import cn.ac.iie.di.base.Startup;
import com.alibaba.fastjson.JSONObject;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import org.apache.log4j.Logger;
import org.apache.commons.io.FileUtils;
/**
 *
 * @author LXHs
 */
public class Writer implements Runnable{
    private static Logger Log=Logger.getLogger(Writer.class.getName());
    
//    private static List<String>msgtype=new  ArrayList<String>();//msg_type可能值，一次读取不再改变
    private static  Map<String,String>msgtype_yulun=new  HashMap<String,String>(10);//舆论场可能值，一次读取不再改变
    
    private static String during_dir;//临时的文件夹，一次读取不再改变
//    private static String target_dir;//目标的文件夹，一次读取不再改变，
    private static int Thread_name;//仅为了标识线程个数
    
    private List<String>during=new ArrayList<String>();//具体由原混合文件生成每个临时文件的位置,最后打印日志用
//    private List<String>target=new ArrayList<String>();//具体由原混合文件生成每个临时文件的位置对应的龙存文件位置，最后打印日志用
    private Map<String,Integer>num=new HashMap<String,Integer>();
    
    //维护msgtype和多个临时文件的映射关系，得到每行msg_type可以立即得到printwriter，写入
    private Map<String,PrintWriter>pwMap=new HashMap<String,PrintWriter>(16);
    private SimpleDateFormat sdf=new SimpleDateFormat("yyyyMMdd");
    static{
        for(String yulun_str:Startup.config.getList("yulun")){
            for(String msg_str:Startup.config.getList("msgtype."+yulun_str)){
                msgtype_yulun.put(msg_str,yulun_str);
            }
        };
//        Log.info("msgtype "+msgtype);
        Log.info("yulun "+msgtype_yulun);
        during_dir=Startup.config.getProperty("during");
//        target_dir=Startup.config.getProperty("target");
    }
    
    
    /**
     * 给 读完一个文件后，要初始化两个map
     * @param path 原混合文件地址
     */
//    public void begin(String path){
//        during.clear();
//        target.clear();
//        for(int i=0;i<msgtype.size();i++){
//            PrintWriter pw = null;
//            String date=null;
//            date=sdf.format(new Date());
//            
//            String sbtarget=add(target_dir,date,yulun.get(i),"text",msgtype.get(i),path);//龙村目录结构：原混合文件夹/日期/舆论场/msgtype/
//            String sbduring=add(during_dir,msgtype.get(i),path);//临时文件结构：临时文件夹/msgtype/源文件名字
//            
//            //确保临时文件存在
//            File dir=new File(add(during_dir,msgtype.get(i)));
//            if(!dir.exists()){
//                dir.mkdirs();
//        }
//            
//            //新临时，龙存的文件位置
//            during.add(sbduring);//最后复制用
//            target.add(sbtarget);//最后复制用
//            
//            //创建每个对应的PrintWriter
//            try{
//                File file=new File(sbduring);
//                pw = new PrintWriter(new OutputStreamWriter(new FileOutputStream(file,true),"utf-8"));
//            } catch (FileNotFoundException ex) {
//                Log.error(ex.getMessage(),ex);
//            } catch (UnsupportedEncodingException ex) {
//                Log.error(ex.getMessage(),ex);
//            }catch (Exception ex) {
//                Log.error(ex.getMessage(),ex);
//            }
//            num.put(msgtype.get(i), 0);
//            pwMap.put(msgtype.get(i), pw);
//        }
//        num.put("all", 0);
//    }
    /**
     * 拼接文件路径用
     * @param sb
     * @param str
     * @return sb/str[0]/str[1]/str[2]...
     */
    private String add(String sb,String... str){
        StringBuilder start=new StringBuilder(sb);
        for(String s:str){
            start.append(File.separator);
            start.append(s);
        }
        return start.toString();
    }
//    public static class Infotime{
//        static Map<Long,Long>map=new HashMap<>();
//        Long start=System.currentTimeMillis();
//        public static void gettime(){
//            Long id=Thread.currentThread().getId();
//            Long start;
//            if(map.get(id)==null){
//                map.put(id,System.currentTimeMillis());
//            }else{
//                System.out.println(System.currentTimeMillis()-map.get(id)+" ms");
//                map.put(id,System.currentTimeMillis());
//            };
//        }
//    }
    
//    public static void main(String arges[]){
//        Infotime.gettime();
//        try {
//            Thread.sleep(1000l);
//        } catch (InterruptedException ex) {
//            java.util.logging.Logger.getLogger(Writer.class.getName()).log(Level.SEVERE, null, ex);
//        }
//        Infotime.gettime();
//    }

    @Override
    public void run() {
        Thread.currentThread().setName("Thread_Writer_"+Thread_name++);
        Log.info("Start_Writer");
        while(!Startup.stop||Startup.queue.size()>0){//只有开关闭，并且队列信息消费完才能结束循环
            String path=null;
            if((path = Startup.queue.poll())==null){//如果没有数据，睡眠5秒，然后重新循环
                try {
                    Log.info("queue is Empty");
                    Thread.sleep(5000l);
                    continue;
                } catch (InterruptedException ex) {
                    Log.error(ex.getMessage(),ex);
                }
            };
            
            BufferedReader br = null ;
            String line=null;
//            begin(path);//建立映射
            pwMap.clear();
            num.clear();
            during.clear();
            num.put("all",0);
            Log.info("begin_aload");
            try {
                br = new BufferedReader(new InputStreamReader(new FileInputStream(new File(Startup.config.getProperty("path")+File.separator+path)),"utf-8"));
                String date=sdf.format(new Date());
                while((line=br.readLine())!=null){
                    Log.debug("duringfile"+line);
                    
                    num.put("all", num.get("all")+1);
                    //拿到与msgtype对应的printwriter,并写入
                    JSONObject json=JSON.parseObject(line);
                    String msg_type=json.get("msg_type").toString();
                    String key=json.get("key").toString();
                    while(msgtype_yulun.get(msg_type)==null){
                        try {
                            Thread.sleep(10000l);
                            Log.error("msg_type error: "+path+" "+msg_type);
                            if(Startup.stop)return;
                        } catch (InterruptedException ex) {
                            Log.error(ex.getMessage(),ex);
                        }
                    }
                    getPw(key,msg_type,path,date).println(line);
                }
            } catch (FileNotFoundException ex) {
                Log.error(ex.getMessage(),ex);
            } catch (IOException ex) {
                Log.error(ex.getMessage(),ex);
            }catch (Exception ex) {
                Log.error(ex.getMessage(),ex);
            }finally{
                Log.info("readfile "+path+" "+ num.get("all"));
                for(PrintWriter pw:pwMap.values()){
                    pw.flush();
                    try{
                        pw.close();
                    }catch(Exception ex){
                        Log.error(ex.getMessage(),ex);
                    }
                }
                
            }
            //关闭
//            Log.debug(msgtype);
//            Log.debug(during);
//            Log.debug(target);
//            for(int i=0;i<msgtype.size();i++){
//                PrintWriter pw=pwMap.get(msgtype.get(i));
//                pw.flush();
//                pw.close();
//                try {
//                    if(new File(during.get(i)).length()==0){
//                        Log.info("targetfile "+path+" "+msgtype.get(i)+" "+0);
//                        continue;
//                    }
//                    FileUtils.copyFile(new File(during.get(i)), new File(target.get(i)));
//                    Log.info("targetfile "+path+" "+msgtype.get(i)+" "+num.get(msgtype.get(i)));
//                } catch (IOException ex) {
//                    Log.error(ex.getMessage(),ex);
//                }catch (Exception ex) {
//                    Log.error(ex.getMessage(),ex);
//                }
//            }
            for(String pw:during){
                try {
                    new File(pw+".ok").createNewFile();
                } catch (IOException ex) {
                    Log.error(ex.getMessage(),ex);
                }
            }
            for(Map.Entry<String,Integer> file:num.entrySet()){
                Log.info("write ： "+file.getKey()+" : "+file.getValue());
            }
//            Log.info(Arrays.toString(new File(Startup.config.getProperty("path")).list()));
//            new File(Startup.config.getProperty("path")+File.separator+path).delete();
//            new File(Startup.config.getProperty("path")+File.separator+path+".ok").delete();
//            new File(Startup.config.getProperty("path")+File.separator+path+".sns").delete();
//            Log.info(Arrays.toString(new File(Startup.config.getProperty("path")).list()));
        }
        Log.info("Writer over");
    }

    private PrintWriter getPw(final String key, final String msg_type,final String path,String date) {
        String key_type=add(key,msg_type);
        if(pwMap.get(key_type)==null){
            String sbduring=add(during_dir,msgtype_yulun.get(msg_type),"text",date,key,msg_type,path);//舆论场/text/日期/key/数据类型/文件
            File dir=new File(add(during_dir,msgtype_yulun.get(msg_type),"text",date,key,msg_type));
            num.put(key_type,1);
            if(!dir.exists()){
                dir.mkdirs();
            }
            PrintWriter pw = null;
            try{
                File file=new File(sbduring);
                pw = new PrintWriter(new OutputStreamWriter(new FileOutputStream(file,true),"utf-8"));
            } catch (FileNotFoundException ex) {
                Log.error(ex.getMessage(),ex);
            } catch (UnsupportedEncodingException ex) {
                Log.error(ex.getMessage(),ex);
            }catch (Exception ex) {
                Log.error(ex.getMessage(),ex);
            }
            during.add(sbduring);
            pwMap.put(key_type, pw);
            return pw;
        }else{
            num.put(key_type, num.get(key_type)+1);
            return pwMap.get(key_type);
        }
    }
}